package com.zhaosc.rx.example;

import java.util.concurrent.TimeUnit;

import org.junit.Test;

import com.google.common.collect.Lists;
import com.zhaosc.rx.utils.SleepUtils;

import io.reactivex.Observable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Predicate;

public class ConditionOperationExample {

	@Test
	public void testAll() {
		Observable.range(1, 100).contains(11)
//		.all(new Predicate<Integer>() {
//
//			@Override
//			public boolean test(Integer t) throws Exception {
//				return t>=-1;
//			}
//		})
				.subscribe(new Consumer<Boolean>() {

					@Override
					public void accept(Boolean t) throws Exception {
						System.out.println(t);
					}
				});

	}

	@Test
	public void testAmb() {
		Observable.ambArray(Observable.just("1", "3", "5"), Observable.just("2", "4", "6"),
				Observable.just("7", "8", "9")).subscribe(new Consumer<String>() {

					@Override
					public void accept(String t) throws Exception {
						System.out.println("********" + t);

					}
				});

	}

	/**
	 * defaultlffimpty和 switchlffimpty 的区别是，
	 * defaultlffimpty操作符只能在被观察者不发送数据时发送一个默认的数据， 如果想要发送更多数据，则可以使用
	 * switch!ffimpty操作符，发送自 定义的被观察者作为替代。
	 */
	@Test
	public void testDefaultEmpty() {
		Observable.<String>empty()
//		.defaultIfEmpty("222")
//		.defaultIfEmpty("333")
				.switchIfEmpty(Observable.just("1", "3", "5")).subscribe(new Consumer<String>() {
					@Override
					public void accept(String t) throws Exception {
						System.out.println("********" + t);

					}
				});

	}

	/**
	 * 传递两个 Observable给 sequenceEqual操作符时， 它会比较两个 Observable
	 * 的发射物，如果两个序列相同(相同的数据，相同的顺序，相同的终止状态)， 则 发射true，否则发射 false
	 */

	@Test
	public void testSequenceEqual() {
		Observable.sequenceEqual(Observable.just("1", "2", "3"), Observable.just("1", "3", "2")).subscribe((t) -> {
			System.out.println(t);
		});

		Observable.sequenceEqual(Observable.just(1, 2, 3), Observable.just(2, 3, 4), (t1, t2) -> {
			System.out.println("t1" + t1 + "****t2:" + t2);
			return t1 < t2;
		}).subscribe((t) -> {
			System.out.println(t);
		});

	}

	/**
	 * skipUntil订阅原始的 Observable，
	 * 但是忽略它的发射物， 直到第二个 Observable发射一项数据那一刻， 
	 * 它才开始发射原始 Observable。 skipUntil默认不在任何特定的调度器上执行
	 * takeUtils类似
	 */
	@Test
	public void testSkipUtil() {

		Observable.intervalRange(0, 110, 0, 1, TimeUnit.MILLISECONDS)
		.skipUntil(Observable.timer(3, TimeUnit.MILLISECONDS))
		.subscribe((t)->{
			System.out.println(t);
		});
		
		SleepUtils.sleep(10000L);
		
	}
	
	/**
	 * skipWhile 订阅原始的 Observable，但是忽略它的发射物 ， 直到指定的某个条件变为 false, 
	 * 它才开始发射原始的 Observable。 skipWhile 默认不在任何特定的调度器上执行 
	 * takeWhile类似
	 */
	@Test
	public void testSkipWhile() {

		Observable.intervalRange(0, 110, 0, 1, TimeUnit.MILLISECONDS).skipWhile((t) -> t<=50).subscribe((t) -> {
			System.out.println(t);
		});

		SleepUtils.sleep(10000L);

	}
}
